package com.wallet.trans.listener;

import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpStatus;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.wallet.trans.dao.SysRocketmqDedupDao;
import com.wallet.trans.dao.WltAddressDao;
import com.wallet.trans.domain.CoinInfo;
import com.wallet.trans.domain.ProducerMqMessage;
import com.wallet.trans.domain.TransMqMessage;
import com.wallet.trans.entity.CallBackReq;
import com.wallet.trans.entity.SysRocketmqDedup;
import com.wallet.trans.entity.TransactionReq;
import com.wallet.trans.entity.WltAddress;
import com.wallet.trans.enums.ResultCodeEnum;
import com.wallet.trans.producer.TransMqProducer;
import com.wallet.trans.resp.SendToResp;
import com.wallet.trans.utils.RedisUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

// 手动应答 https://www.jianshu.com/p/501d66af4f20
@Component
@Slf4j
@RocketMQMessageListener(consumerGroup = "${trans.group}",
        topic = "${trans.topic}")
public class TransMqConsumer implements RocketMQListener<MessageExt> {

    @Value("${coin-service.send-to-url}")
    private String sendToUrl;
    @Value("${coin-service.call-back-url}")
    private String callBackUrl;
    @Value("${trans.business_id}")
    private Integer businessId;
    @Value("${trans.wallet_id}")
    private Integer walletId;
    @Value("${trans.group}")
    private  String group;
    //    @Autowired
//    private WltAddressDao wltAddressDao;
    @Autowired
    private RedisUtils redisUtils;

    @Autowired
    private TransMqProducer transMqProducer;

    @Autowired
    private SysRocketmqDedupDao sysRocketmqDedupDao;


    @Override
    public void onMessage(MessageExt message) {

        Map<String, String> heads = new HashMap<>();
        heads.put("Content-Type", "application/json;charset=UTF-8");

        TransMqMessage transMqMessage = JSON.parseObject(message.getBody(), TransMqMessage.class);
        log.info("收到消息message【{}】transMqMessage 【{}】 准备出金", message, transMqMessage);
        log.info("msgId【{}】", message.getMsgId());
        LambdaQueryWrapper<SysRocketmqDedup> queryWrapper = Wrappers.lambdaQuery();
        queryWrapper.eq(SysRocketmqDedup::getMsgId, message.getMsgId().toString());
        SysRocketmqDedup sysRocketmqDedupMsg = sysRocketmqDedupDao.selectOne(queryWrapper);
        if(Objects.isNull(sysRocketmqDedupMsg)){
            SysRocketmqDedup sysRocketmqDedup = new SysRocketmqDedup();
            sysRocketmqDedup.setMqGroup(group);
            sysRocketmqDedup.setMqTopic(message.getTopic());
            sysRocketmqDedup.setSerialNo(transMqMessage.getSerialNo());
            sysRocketmqDedup.setMsgId(message.getMsgId().toString());
            sysRocketmqDedup.setMsgBody(JSONObject.toJSONString(transMqMessage));
            sysRocketmqDedup.setCreateTime(System.currentTimeMillis());
            sysRocketmqDedupDao.insert(sysRocketmqDedup);
            log.info("插入记录【{}】", sysRocketmqDedup);

            ProducerMqMessage producerMqMessage = new ProducerMqMessage();
            producerMqMessage.setMessage(transMqMessage);
            producerMqMessage.setCode(ResultCodeEnum.SYSTEM_ERROR.getCode());
            producerMqMessage.setMsg(ResultCodeEnum.SYSTEM_ERROR.getMessage());

            //从from获取商户ID
//        LambdaQueryWrapper<WltAddress> queryWrapper = Wrappers.lambdaQuery();
//        queryWrapper.eq(WltAddress::getAddress, message.getFromAddress());
//        WltAddress wltAddress = wltAddressDao.selectOne(queryWrapper);
//        if (Objects.isNull(wltAddress)) {
//            producerMqMessage.setCode(ResultCodeEnum.ADDRESS_ERROR.getCode());
//            producerMqMessage.setMsg(ResultCodeEnum.ADDRESS_ERROR.getMessage());
//            log.error("未找到地址对应的商户ID,message【{}】", JSONObject.toJSON(message));
//        } else {
            //获取币种信息
            Object obj = redisUtils.get("waas:cloudAdmin:coinInfoRedisKey-" + transMqMessage.getSymbol());
            CoinInfo coinInfo = JSON.parseObject(obj.toString(), CoinInfo.class);
            if (Objects.isNull(coinInfo)) {
                producerMqMessage.setCode(ResultCodeEnum.SYMBOL_ERROR.getCode());
                producerMqMessage.setMsg(ResultCodeEnum.SYMBOL_ERROR.getMessage());
                log.error("未找到币种信息,message【{}】", JSONObject.toJSON(transMqMessage));
                sysRocketmqDedup.setStatus(0);
            } else {
                //构建sendto需要参数
                TransactionReq transactionReq = new TransactionReq();
                transactionReq.setAppId(walletId);
                transactionReq.setBusinessId(businessId);
                transactionReq.setSymbol(transMqMessage.getSymbol());
                transactionReq.setFrom(transMqMessage.getFromAddress());
                transactionReq.setTo(transMqMessage.getToAddress());
                transactionReq.setAmount(transMqMessage.getAmount());
                transactionReq.setContract(coinInfo.getContractAddress());
                transactionReq.setMemo(transMqMessage.getMemo());
                transactionReq.setForceSend(transMqMessage.getForceSend());
                //发送sendto
                log.info("发送到coinservice 参数{}", JSONObject.toJSON(transactionReq));
                HttpResponse execute = HttpRequest.post(sendToUrl).headerMap(heads, false).body(JSONUtil.toJsonStr(transactionReq)).execute();
                if (execute.getStatus() == HttpStatus.HTTP_OK) {
                    SendToResp bean = JSONUtil.toBean(execute.body(), SendToResp.class);
                    if (Objects.equals(bean.getCode(), ResultCodeEnum.PASS.getCode())) {
                        CallBackReq callBackReq = new CallBackReq();
                        callBackReq.setHash(bean.getData().getHash());
                        callBackReq.setHex(bean.getData().getHex());
                        callBackReq.setTransMqMessage(transMqMessage);
                        callBackReq.setSerialNo(transMqMessage.getSerialNo());
                        //成功后回调
//                    HttpResponse execute1 = HttpRequest.post(callBackUrl).headerMap(heads, false).body(JSONUtil.toJsonStr(callBackReq)).execute();
                        //只要http状态是200就认为成功
//                    if (execute1.getStatus() == 200) {
//                        log.info("回调成功:{}",message);
                        producerMqMessage.setData(bean.getData());
                        producerMqMessage.setCode(ResultCodeEnum.PASS.getCode());
                        producerMqMessage.setMsg(ResultCodeEnum.PASS.getMessage());
//                    } else {
//                        log.error("回调请求失败:{}", execute1);
//                    }
                        if(Objects.isNull(bean.getData().getHash()) && Objects.nonNull(bean.getData().getHex()))
                            sysRocketmqDedup.setTxKey(bean.getData().getHex());
                        sysRocketmqDedup.setStatus(1);
                    } else {
                        producerMqMessage.setCode(bean.getCode());
                        producerMqMessage.setMsg(bean.getMsg());
                        sysRocketmqDedup.setStatus(0);
                        log.error("请求coinservice 业务失败,结果:{}", JSONObject.toJSON(bean));
                    }
                } else {
                    //如果是coinservice网络错误，进行重试
                    log.error("请求coinservice http状态错误,resp:{}", JSONObject.toJSON(execute));
                    sysRocketmqDedup.setStatus(0);
                    throw new RuntimeException("请求coinservice http状态错误");
                }
            }
//        }
            sysRocketmqDedup.setResp(JSONObject.toJSONString(producerMqMessage));
            sysRocketmqDedupDao.updateById(sysRocketmqDedup);
            transMqProducer.send(producerMqMessage);
            log.info("执行完成了");
        }
    }
}